package com.etc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;

/**
 * @Auther: zhezhe
 * @Date: 2018/10/2914:29
 * @Description:
 */
public class DataMaxMinAvg {

    public static class MapData extends Mapper<LongWritable, Text, IntWritable,IntWritable>{
        /**
         *
         * @param key  为1
         * @param value
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */

        @Override
        protected void map(LongWritable key, Text value, Context context) throws
                IOException, InterruptedException {

            String line = value.toString();
            String[] s = line.split(" ");

        for (String s1 : s) {
                context.write(new IntWritable(1),new IntWritable(Integer.parseInt(s1)));
              //  System.out.println(num+s1);
            }
        }
    }

        public static class ReduceData extends Reducer<IntWritable,IntWritable,Text,IntWritable>{
            int max = -65536;
            int min = 65535;
            int sum = 0;
            int num = 0;

            @Override
            protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

                for (IntWritable value : values) {
                    if (value.get()>max){
                        max = value.get();
                    }
                    if (value.get()<min){
                        min = value.get();
                    }
                    num++;
                    sum = sum +value.get();
                }
                context.write(new Text("最大值为："),new IntWritable(max));
                context.write(new Text("最小值为："),new IntWritable(min));
                context.write(new Text("和为："),new IntWritable(sum));
                context.write(new Text("平均数为："),new IntWritable(sum/num));
            }
        }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //读取配置文件
        Configuration conf = new Configuration();
        //创建一个工作
        Job job = Job.getInstance(conf);

        job.setJarByClass(DataMaxMinAvg.class);
        job.setMapperClass(MapData.class);
        job.setReducerClass(ReduceData.class);

        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Integer.class);

        //如果输出路径已经存在，则将其删除
        File file = new File("E:\\mapreduce取最值平均数\\数据\\output");
        if (file.exists()){

            FileUtil.fullyDelete(file);
        }
        // 指定要处理的原始数据存放的路径
        FileInputFormat.addInputPath(job,new Path("E:\\mapreduce取最值平均数\\数据\\input\\*"));
        // 指定处理之后的结果输出到哪个路径
        FileOutputFormat.setOutputPath(job,new Path("E:\\mapreduce取最值平均数\\数据\\output"));

        //
        job.setNumReduceTasks(1);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
